package org.example.mq.mqserver.datacenter;

import org.example.mq.common.exception.MqException;
import org.example.mq.mqserver.core.*;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.file.Path;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;

/*
    增查删
 */

public class MemoryDataCenter {
    //交换机管理 <exchangeName,Exchange>
    ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();

    //队列管理  <QueueName,MessageQueue>
    ConcurrentHashMap<String, MessageQueue>messageQueueMap = new ConcurrentHashMap<>();

    //消息管理
    ConcurrentHashMap<String, Message>messageMap = new ConcurrentHashMap<>();

    //绑定关系管理 <ExchangeName,<QueueName,Set<BindingKey>>>
    ConcurrentHashMap<String,ConcurrentHashMap<String, Set<String>>> bindingMap = new ConcurrentHashMap<>();

    //队列和消息的关系集合
    ConcurrentHashMap<String, LinkedBlockingQueue<Message>>queueMessageMap = new ConcurrentHashMap<>();

    //消息ack管理 <queueName,<messageID,message>>   死信队列
    ConcurrentHashMap<String,ConcurrentHashMap<String,Message>> queueAckMap = new ConcurrentHashMap<>();

    //-----------------交换机-----------------
    public void insertExchange(Exchange exchange){
        exchangeMap.put(exchange.getName(),exchange);
    }

    public Exchange getExchange(String exchangeName){
        return exchangeMap.get(exchangeName);
    }

    public boolean delExchange(String exchangeName){
        return exchangeMap.remove(exchangeName)!=null;
    }


    //-----------------队列------------------
    public void insertQueue(MessageQueue messageQueue){
        messageQueueMap.put(messageQueue.getName(),messageQueue);
    }

    public MessageQueue getQueue(String queueName){
        return messageQueueMap.get(queueName);
    }

    public boolean delQueue(String queueName){
        return messageQueueMap.remove(queueName)!=null;
    }

    //----------------消息--------------------
    public void addMessage(Message message){
        messageMap.put(message.getMessageId(),message);
    }

    public Message getMessage(String messageId){
        return messageMap.get(messageId);
    }

    public boolean delMessage(String messageId){
        return messageMap.remove(messageId)!=null;
    }

    //----------------绑定--------------------
    public void insertBinding(Bind bind){
        Exchange exchange = getExchange(bind.getExchangeName());

        //确保交换机和队列都存在
        if(exchange==null || !messageQueueMap.containsKey(bind.getQueueName())){
            throw new MqException("[MemoryDataCenter]insertBinding:新增bind失败,交换机或者队列不存在!");
        }

        //判断exchange的bind是否提前存在
        bindingMap.computeIfAbsent(bind.getExchangeName(),t->new ConcurrentHashMap<>());

        //判断bindKey是否重复
        Set<String> bindKeys = bindingMap.get(bind.getExchangeName()).computeIfAbsent(bind.getQueueName(), k -> ConcurrentHashMap.newKeySet());
        //如果交换机类型是fanout 且bindingKey为 ""那么就跳过
        if(exchange.getType()!=ExchangeType.FANOUT && !bind.getBindingKey().isEmpty()){
            if (!bindKeys.contains(bind.getBindingKey())) {
                bindKeys.add(bind.getBindingKey());
            } else {
                throw new MqException("[MemoryDataCenter]insertBind:交换机:" + bind.getExchangeName() +
                        ",队列:" + bind.getQueueName() + "，bindKey重复");
            }
        }
    }

    //根据exchangeName和QueueName 来获取bind
    public Set<String> getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Set<String>> queueMap = bindingMap.get(exchangeName);
        if (queueMap == null) {
            return null;
        }
        Set<String> bindingKeys = queueMap.get(queueName);
        if (bindingKeys == null) {
            return null;
        }
        return bindingKeys;
    }


    //根据exchangeName 获取全部bind
    public ConcurrentHashMap<String,Set<String>> getBindings(String exchange){
        return bindingMap.get(exchange);
    }

    public boolean delBind(Bind bind){
        //检查是否有bind关系
        ConcurrentHashMap<String, Set<String>> binds = bindingMap.get(bind.getExchangeName());
        if(binds==null || binds.isEmpty()){
            throw new MqException("[Exchange:" + bind.getExchangeName() + "],并未绑定交换机");
        }

        Set<String> bindKeys = binds.get(bind.getQueueName());

        if(bindKeys==null || bindKeys.isEmpty()){
            throw new MqException("[Exchange:"+bind.getExchangeName() + "]与[Queue:"+bind.getQueueName()+"],并无绑定关系");
        }

        return bindKeys.remove(bind.getBindingKey());
    }

    //根据queueName 获取对应bind
    //TODO

    //删除exchange所有bind
    public ConcurrentHashMap<String, Set<String>> delExchangeBind(String exchangeName){
        if(!exchangeMap.containsKey(exchangeName)){
            return null;
        }
        ConcurrentHashMap<String, Set<String>> remove = bindingMap.remove(exchangeName);
        return remove;
    }

    //---------------队列消息关系---------------
    //发送消息到队列中
    public void sendMessage(MessageQueue messageQueue,Message message) throws InterruptedException {
        //判断队列是否存在
        if(!messageQueueMap.containsKey(messageQueue.getName())){
            throw new MqException("[MemoryDataCenter]sendMessage:发送的指定队列不存在！");
        }

        LinkedBlockingQueue<Message> messages = queueMessageMap.computeIfAbsent(messageQueue.getName(), f -> new LinkedBlockingQueue<>(Integer.MAX_VALUE));
        messages.put(message);
        addMessage(message);
    }

    //消息队列中取出消息
    public Message takeMessageFromMessageQueue(String queueName) throws InterruptedException {
        //验证队列
        if(!messageQueueMap.containsKey(queueName)){
            throw new MqException("[MemoryDataCenter]takeMessage:取消息的指定队列不存在！");
        }

        LinkedBlockingQueue<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return null;
        }

        return messages.poll();
    }

    //获取消息队列中消息数量
    public int getMessageCount(String queueName){
        //用于监控或展示都很好。只要不用于消费决策逻辑就可以放心用。
        LinkedBlockingQueue<Message> messages = queueMessageMap.get(queueName);
        if(messages==null)return 0;

        return messages.size();
    }

    //---------------消息ack---------------

    //添加未确认消息
    public void addMessageWaitAck(String queueName,Message message){
        queueAckMap.computeIfAbsent(queueName,f->new ConcurrentHashMap<>()).put(message.getMessageId(),message);
    }
    //删除未确认消息
    public void delAckMessage(String queueName,String messageId){
        ConcurrentHashMap<String, Message> messages = queueAckMap.get(queueName);
        if(messages==null)return;
        messages.remove(messageId);
    }
    //获取指定未确认消息
    public Message getWaitAckMessage(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> ackMessages = queueAckMap.get(queueName);
        if (ackMessages == null) {
            return null;
        }
        return ackMessages.get(messageId); // 线程安全
    }

    //内存数据初始化（从数据库中读取数据到内存中）
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, InterruptedException {
        //交换机恢复
        List<Exchange> exchanges = diskDataCenter.queryAllExchange();
        exchanges.forEach(this::insertExchange);
        //队列恢复
        List<MessageQueue> messageQueues = diskDataCenter.queryAllMessageQueue();
        for (MessageQueue messageQueue : messageQueues) {
            insertQueue(messageQueue);
        }
        //绑定恢复
        List<Bind> binds = diskDataCenter.queryAllBind();
        binds.forEach(this::insertBinding);
        //消息恢复
        for (MessageQueue messageQueue : messageQueues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(messageQueue.getName());
            for (Message message : messages) {
                sendMessage(messageQueue,message);
                addMessage(message);
            }
        }
    }
}
